/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */


package org.apache.skywalking.apm.plugin.asf.dubbo;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
import java.lang.reflect.Method;
import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
import org.apache.skywalking.apm.agent.core.context.tag.Tags;
import org.apache.skywalking.apm.agent.core.context.CarrierItem;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;

/**
 * {@link DubboInterceptor} define how to enhance class {@link org.apache.dubbo.monitor.support.MonitorFilter#invoke(Invoker,
 * Invocation)}. the trace context transport to the provider side by {@link RpcContext#attachments}.but all the version
 * of dubbo framework below 2.8.3 don't support {@link RpcContext#attachments}, we support another way to support it.
 *
 */
public class DubboInterceptor implements InstanceMethodsAroundInterceptor {
    /**
     * <h2>Consumer:</h2> The serialized trace context data will
     * inject to the {@link RpcContext#attachments} for transport to provider side.
     * <p>
     * <h2>Provider:</h2> The serialized trace context data will extract from
     * {@link RpcContext#attachments}. current trace segment will ref if the serialize context data is not null.
     * <p>
     * <p>
     *
     * beforeMethod() 方法会判断当前处于 Consumer 端还是 Provider 端：
     *
     *      如果处于 Consumer 端，则会将当前 TracingContext 上下文序列化成 ContextCarrier 字符串，并填充到 RpcContext 中。
     *      RpcContext 中携带的信息会在之后随 Dubbo 请求一起发送出去，相应的，还会创建 ExitSpan。
     *
     *      如果处于 Provider 端，则会从请求中反序列化 ContextCarrier 字符串，并填充当前 TracingContext 上下文。相应的，创建 EntrySpan。
     *
     *      《Dubbo插件处理逻辑.png》
     */
    @Override
    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
        // invoke()方法的两个参数
        Invoker invoker = (Invoker)allArguments[0];
        Invocation invocation = (Invocation)allArguments[1];

        // RpcContext是Dubbo用来记录请求上下文信息的对象
        RpcContext rpcContext = RpcContext.getContext();

        // 检测当前服务是Consumer端还是Provider端
        boolean isConsumer = rpcContext.isConsumerSide();
        URL requestURL = invoker.getUrl();

        AbstractSpan span;

        final String host = requestURL.getHost();
        final int port = requestURL.getPort();

        if (isConsumer) { // 检测是否为 Consumer
            final ContextCarrier contextCarrier = new ContextCarrier();
            // 如果当前是Consumer侧，则需要创建ExitSpan对象，其中EndpointName是由请求URL地址、服务名以及方法名拼接而成的
            span = ContextManager.createExitSpan(generateOperationName(requestURL, invocation), contextCarrier, host + ":" + port);
            // invocation.getAttachments().put("contextData", contextDataStr);
            // @see htt ps://github.com/alibaba/dubbo/blob/dubbo-2.5.3/dubbo-rpc/dubbo-rpc-api/src/main/java/com/alibaba/dubbo/rpc/RpcInvocation.java#L154-L161
            // 创建CarrierItem链表，其中会根据当前Agent支持的版本号对ContextCarrier进行序列化
            CarrierItem next = contextCarrier.items(); // 调用此方法时会进行序列化
            while (next.hasNext()) {
                next = next.next();
                // 将ContextCarrier字符串填充到RpcContext中，后续会随Dubbo请求一起发出
                rpcContext.getAttachments().put(next.getHeadKey(), next.getHeadValue());
            }
        } else { // 如果当前是Provider侧
            ContextCarrier contextCarrier = new ContextCarrier();
            // 创建CarrierItem链表
            CarrierItem next = contextCarrier.items();
            while (next.hasNext()) {
                next = next.next();
                // 从RpcContext中获取ContextCarrier字符串反序列化，并填充当前上面创建的空白ContextCarrier对象
                next.setHeadValue(rpcContext.getAttachment(next.getHeadKey()));
            }

            // 创建 EntrySpan
            span = ContextManager.createEntrySpan(generateOperationName(requestURL, invocation), contextCarrier);
        }

        // 设置Tags
        Tags.URL.set(span, generateRequestURL(requestURL, invocation));
        // 设置 component
        span.setComponent(ComponentsDefine.DUBBO);
        // 设置 SpanLayer
        SpanLayer.asRPCFramework(span);
    }

    // 检查请求结果是否有异常，如果有异常，则通过 Log 将异常的堆栈信息记录到当前 Span 中，并在当前 Span 设置异常标志（即 errorOccurred 字段设置为 true）
    @Override
    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, Object ret) throws Throwable {
        Result result = (Result)ret;
        if (result != null && result.getException() != null) {
            dealException(result.getException());
        }

        // 调用 ContextManager.stopSpan() 方法关闭当前 Span（也就是 beforeMethod() 方法中创建的 EntrySpan 或 ExitSpan）
        ContextManager.stopSpan();
        return ret;
    }

    @Override
    public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, Throwable t) {
        dealException(t);
    }

    /**
     * Log the throwable, which occurs in Dubbo RPC service.
     */
    private void dealException(Throwable throwable) {
        AbstractSpan span = ContextManager.activeSpan();
        span.errorOccurred();
        span.log(throwable);
    }

    /**
     * Format operation name. e.g. org.apache.skywalking.apm.plugin.test.Test.test(String)
     *
     * @return operation name.
     */
    private String generateOperationName(URL requestURL, Invocation invocation) {
        StringBuilder operationName = new StringBuilder();
        operationName.append(requestURL.getPath());
        operationName.append("." + invocation.getMethodName() + "(");
        for (Class<?> classes : invocation.getParameterTypes()) {
            operationName.append(classes.getSimpleName() + ",");
        }

        if (invocation.getParameterTypes().length > 0) {
            operationName.delete(operationName.length() - 1, operationName.length());
        }

        operationName.append(")");

        return operationName.toString();
    }

    /**
     * Format request url.
     * e.g. dubbo://127.0.0.1:20880/org.apache.skywalking.apm.plugin.test.Test.test(String).
     *
     * @return request url.
     */
    private String generateRequestURL(URL url, Invocation invocation) {
        StringBuilder requestURL = new StringBuilder();
        requestURL.append(url.getProtocol() + "://");
        requestURL.append(url.getHost());
        requestURL.append(":" + url.getPort() + "/");
        requestURL.append(generateOperationName(url, invocation));
        return requestURL.toString();
    }
}
